package com.explorex.infra.puma.gateway.netty.common;

import com.google.common.collect.Lists;
import lombok.AccessLevel;
import lombok.Data;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

@Slf4j
@Data
public class BaseCollector<T, E> implements ApplicationListener<ContextClosedEvent> {

    private ExecutorService pool;

    @Setter(AccessLevel.NONE)
    private LinkedBlockingQueue<T> queue = new LinkedBlockingQueue<>(); // enQueue T

    private AggregateHandler<T, E> aggHandler;  // aggregate multi of T to E

    private CustomHandler<E> batchHandler; // costume E(aggregated T)

    @Setter(AccessLevel.NONE)
    private volatile boolean stop = false;

    private long perTs = 100;

    private int perCount = 100;

    // 若为true, 则每批次都要等待perTs; 否则，若perCount满，不等待
    private boolean forceWait = false;

    // 泛型T的class simpleName, 在上报时会用到
    private String elementName;

    private int warmUpNum = 10;

    @Resource
    private ExecutorService executorService;


    private ScheduledExecutorService scheduledExecutor = new ScheduledThreadPoolExecutor(1, new BasicThreadFactory.Builder().namingPattern(this.getClass().getSimpleName() + "Schedule-%d").build());

    public BaseCollector(ExecutorService pool, CustomHandler<E> handler, AggregateHandler<T, E> aggHandler, int perTs, int perCount, boolean forceWait, String elementName) {
        this.pool = pool;
        this.batchHandler = handler;
        this.aggHandler = aggHandler;
        this.perTs = perTs;
        this.perCount = perCount;
        this.forceWait = forceWait;
        this.elementName = elementName;
    }

    @PostConstruct
    public void init() throws Exception {
        // 必要项检查
        if (pool == null || batchHandler == null || ( elementName == null)) {
            throw new Exception("init fail for pool, batchHandler or elementName not initialized");
        }

        // 每隔1s上报队列长度, 长度存于field len中, 可在influxdb上直接查看sum(len)
        // 预热线程池
        warmUpPool(warmUpNum);

        // 单线程从消息队列中取数, 再通过consumeQueue提交到 pool线程池 执行具体业务逻辑(handler实现业务逻辑)
        executorService.submit(() -> {
            while (!stop) {
                try {
                    long begin = System.nanoTime();
                    int count = consumeQueue();
                    long costTs = (System.nanoTime() - begin) / 1000000;// 纳秒转毫秒
                    // 队列未满, perTs处理一次; 否则一直处理
                    if (forceWait || count < perCount) {
                        long sleep = costTs >= perTs ? 0 : perTs - costTs;
                        if (sleep > 0) {
                            Thread.sleep(sleep);
                        }
                    }
                } catch (Exception e) {
                    log.error("consumeQueue exception {}", elementName, e);
                }
            }
        });
        log.info("{} inited config perTs {} perCount {} forceWait {}", this.getClass().getSimpleName(), perTs, perCount, forceWait);
    }

    public void enQueue(T data) {
        if (!stop) {
            queue.offer(data);
        }
    }

    private int consumeQueue() {
        List<T> dataList = Lists.newArrayList();
        try {
            T firstData = queue.take();  //take will wait until element becomes available, 避免空转
            dataList.add(firstData);
        } catch (InterruptedException e) {
            return 0;
        }
        int count = 1;  // firstData has one
        T data;
        while ((data = queue.poll()) != null) {
            dataList.add(data);
            count++;
            if (count == perCount)
                break;
        }
        pool.submit(() -> {
            try {
                batchHandler.run(aggHandler.agg(dataList));
            } catch (Exception e) { // 兜底多线程的异常，打印出来方便定位
                log.error("consumeQueue throw exception {}", dataList, e);

            } finally {
                int size = dataList.size();
            }
        });
        return count;
    }

    private void warmUpPool(int num) {
        List<IdleCallable> warmupTasks = Lists.newArrayList();
        for (int i = 0; i < num; i++)
            warmupTasks.add(new IdleCallable());
        try {
            pool.invokeAll(warmupTasks, 1, TimeUnit.SECONDS);
            log.info("warm up {} pool thread num {}", this.getClass().getSimpleName(), num);
        } catch (InterruptedException e) {
            log.error("warmUp {} failed", this.getClass().getSimpleName());
        }
    }

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        // stop put element into queue
        stop = true;
        log.info("{} destroy start, left queue size {}", this.getClass().getSimpleName(), queue.size());
        // try to clear queue element before terminate
        int clearSum = 0;
        while (!queue.isEmpty()) {
            clearSum += consumeQueue();
            log.info("{} destroying, total clear queue size {}", this.getClass().getSimpleName(), clearSum);
        }
        // shutdown pool
        if (pool != null && !pool.isShutdown())
            pool.shutdown();
    }

    public interface AggregateHandler<T, E> {
        E agg(List<T> item);
    }

    public interface CustomHandler<E> {
        void run(E item);
    }

    class IdleCallable implements Callable<Integer> {
        @Override
        public Integer call() throws Exception {
            long ts = System.currentTimeMillis();
            return Objects.hashCode(ts);
        }
    }
}

